package org.example.partitioner2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @ClassName FlowDriver
 *
 * 设置分区操作、源码原理也是用的这个
 * @Author 小坏
 * @Date 2021/7/16、18:13
 * @Version 1.0
 */
public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1、获取job
        Configuration config = new Configuration();
        Job job = Job.getInstance(config);

        //2、设置jar
        job.setJarByClass(FlowDriver.class);

        //3、关联mapper 和 reducer
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

        //4、设置mapper 输出的key和value 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //5、设置最终数据输出的key和value 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        //###############################加上这两个配置
        job.setPartitionerClass(ProvincePartitioner.class); //跟自己写的建立连接
        //TODO 1、ReduceTasks 大于 我自定义的ProvincePartitioner 结果数、则会多产生几个空的输出文件part-r-000xx
        //TODO 2、如果1<ReduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会Exception
        //TODO 3、如 果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个 ReduceTask，最终也就只会产生一个结果文件 part-r-00000；
        //TODO 4、分区号必须从零开始，逐一累加。
        /**
         * 例如：假设自定义分区数为5，则
         * （1）job.setNumReduceTasks(1);      会正常运行，只不过会产生一个输出文件
         * （ 2）job.setNumReduceTasks(2);    会报错
         * （3）job.setNumReduceTasks(6);      大于5，程序会正常运行，会产生空文件
         *
         *
         */
        job.setNumReduceTasks(5); //设置文本数5个

             //###############################
        //6、设置数据的输入和输出路径
        FileInputFormat.setInputPaths(job, new Path("D:\\input\\inputflow"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\output09"));

        //7、提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);

    }
}
